package data.platform.timescale.persistence.repository;

import com.fasterxml.jackson.databind.ObjectMapper;
import data.platform.timescale.internal.cache.TsCacheService;
import data.platform.timescale.persistence.mapping.RulePointEO;
import io.r2dbc.spi.Batch;
import io.r2dbc.spi.Row;
import io.r2dbc.spi.RowMetadata;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.r2dbc.core.DatabaseClient;
import org.springframework.stereotype.Repository;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;
import java.util.Set;
import java.util.function.BiFunction;

@ConditionalOnBean(name = "timeScaleConfig")
@Slf4j
@Repository
@RequiredArgsConstructor
public class TsRulePointRepository {

    final DatabaseClient databaseClient;

    private final static String CREATE_SQL = "INSERT INTO rule_point(rule_id, event_time, hit_number, hit_data) values('%s', '%s', %d, '%s')";

    private final static String QUERY_SQL = "SELECT * FROM rule_point WHERE rule_id = :ruleId AND event_time >= :beginTime AND event_time <= :endTime ORDER BY event_time asc";

    private static final ThreadLocal<SimpleDateFormat> dateFormatter = ThreadLocal.withInitial(() -> new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS"));

    public static final BiFunction<Row, RowMetadata, RulePointEO> MAPPING_FUNCTION = (row, rowMetaData) -> RulePointEO.builder()
            .ruleId(row.get("rule_id", String.class))
            .eventTime(row.get("event_time", Date.class))
            .hitNumber(row.get("hit_number", Integer.class))
            .hitData(row.get("hit_data", String.class))
            .build();

    public Mono<Integer> saveBatch(Set<RulePointEO> eos) {
        return databaseClient.inConnection(connection -> {
            Batch batch = connection.createBatch();
            for (RulePointEO eo : eos) {
                String event_time = dateFormatter.get().format(eo.getEventTime());
                String sql = String.format(CREATE_SQL, eo.getRuleId(), event_time, eo.getHitNumber(), eo.getHitData());
                batch.add(sql);
            }
            return Mono.from(batch.execute())
                    .map(result -> eos.size())
                    .doOnNext(id -> {
                        if (log.isDebugEnabled()) {
                            log.debug("create dataPoint size is: [{}].", id);
                        }
                    });
        });
    }

    public Flux<RulePointEO> queryRulePointEO(String ruleId, Date beginTime, Date endTime) {
        return databaseClient.sql(QUERY_SQL)
                .bind("ruleId", ruleId)
                .bind("beginTime", beginTime)
                .bind("endTime", endTime)
                .map(MAPPING_FUNCTION)
                .all();
    }


}
